package com.bd08.flink.demo.zhang

import org.apache.flink.api.scala.{ExecutionEnvironment, createTypeInformation}
import org.apache.flink.core.fs.FileSystem

object Customers extends App {
  case class Customer(id:String,name:String,address:String)
  val env = ExecutionEnvironment.getExecutionEnvironment
  val csvInput =env.readCsvFile[Customer]("H:\\Module 14 - 项目部署\\demo1\\src\\main\\resources\\zhang\\customers.csv",includedFields = Array(1,2,6),quoteCharacter = '"')
  csvInput
    .filter(cusotmer => cusotmer.address.contains("Chicago"))
    .writeAsCsv("H:\\Module 14 - 项目部署\\demo1\\src\\main\\resources\\zhang\\Chicago","\n",",",FileSystem.WriteMode.OVERWRITE)

  csvInput.groupBy("name")
    .reduceGroup(group =>{
      val grp = group.next()
      (grp.name+"|"+grp.address ,group.size)
    })
    .writeAsCsv("H:\\Module 14 - 项目部署\\demo1\\src\\main\\resources\\zhang\\tongming","\n",",",FileSystem.WriteMode.OVERWRITE)
  csvInput.groupBy("name","address")
    .reduceGroup(group =>{
      (group.next().name,group.size)
    })
    .writeAsCsv("H:\\Module 14 - 项目部署\\demo1\\src\\main\\resources\\zhang\\tongming2","\n",",",FileSystem.WriteMode.OVERWRITE)

  env.execute()
}
